package org.example.watermark;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.example.aggres.Dog;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * rich function
 */
public class DataSinkWatermark {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Dog> source = env.fromElements(
                        new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1),
                        new Dog("bb", "b", 5), new Dog("bb", "b", 5), new Dog("bb", "b", 5),
                        new Dog("cc", "b", 6), new Dog("cc", "b", 6),
                        new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8),
                        new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4),
                        new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4)

                )
                /*无序流*/
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Dog>forBoundedOutOfOrderness(
                        Duration.ofSeconds(5)))
                /*有序流*/
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Dog>forMonotonousTimestamps()
                        .withTimestampAssigner((dog, l) -> dog.getAge() * 1000));


        env.setParallelism(1);
//        设置周期性生成watermark
        env.getConfig().setAutoWatermarkInterval(200);
        StreamingFileSink<String> fileSink = StreamingFileSink.<String>forRowFormat(
                        new Path("log1"),
                        new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                                .build()
                )
                .build();
        source.map(data -> data.toString()).addSink(fileSink);

        env.execute();
    }
}


// batchinsert